package com.abyss.sink.stream;


import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;

public class CommitToMysqlBatchDemo extends Thread{
    private Queue<SinkToMysqlDemo.UserInfo> queue;
    private Connection connection;
    private PreparedStatement ps;
    private boolean isRun = true;
    private List<SinkToMysqlDemo.UserInfo> list = new ArrayList<>();
    public CommitToMysqlBatchDemo(Queue queue) throws SQLException {
        this.queue = queue;
        String url = "jdbc:mysql://node1:3306/flinkdemo?useUnicode=true&characterEncoding=utf-8&useSSL=false";
        this.connection = DriverManager.getConnection(url, "root", "123456");
        this.ps = this.connection.prepareStatement("INSERT INTO user VALUES(?, ?, ?, ?)");
    }



    @Override
    public void run() {
        super.run();
        while (isRun) {
            SinkToMysqlDemo.UserInfo userInfo = queue.poll();
            if (userInfo == null) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                continue;
            }
            list.add(userInfo);

            if (list.size() >= 100) {
                for (SinkToMysqlDemo.UserInfo ele: list) {
                    try {
                        ps.setInt(1, ele.getId());
                        ps.setString(2, ele.getUsername());
                        ps.setString(3, ele.getPassword());
                        ps.setString(4, ele.getName());
                    } catch (SQLException throwables) {
                        throwables.printStackTrace();
                    }
                }
                try {
                    ps.executeBatch();
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }

                list.clear();
            }
        }
    }
}